package com.ropa.java;

import java.util.concurrent.*;

/**
 * 阻塞队列的相关问题
 *
 * @author pat
 * @date 2022/04/22
 */
public class BlockingQueuesIssue {
    public static final Class<?>[] blockingQueueClasses = {
            ArrayBlockingQueue.class,
            LinkedBlockingQueue.class,
            LinkedBlockingDeque.class,// 接口为 BlockingDeque
            LinkedTransferQueue.class,// 接口为 TransferQueue
            SynchronousQueue.class, //
            PriorityBlockingQueue.class,// 优先队列
            DelayQueue.class,//延迟队列
    };


    public static void main(String[] args) throws InterruptedException {
//        synchronousQueueDemo();
//        linkedTransferQueue();
        delayQueueDemo();
    }


    /**
     * 同步队列演示
     * <p>
     * SynchronousQueue :
     * SynchronousQueue是线程安全的。
     * SynchronousQueue和其他的BlockingQueue不同的是SynchronousQueue的capacity是0。
     * 即SynchronousQueue不存储任何元素。
     */
    public static void synchronousQueueDemo() throws InterruptedException {

        SynchronousQueue<Object> sq = new SynchronousQueue<>();
        Thread produce = new Thread(() -> {
            try {
                System.out.println("before produce");
                System.out.println("send 123");
                sq.put("123");
                System.out.println("after produce");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        Thread consume = new Thread(() -> {
            try {
                System.out.println("before consume");
                Object take = sq.take();
                System.out.println("got : " + take);
                System.out.println("after consume");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        produce.start();
        consume.start();
        produce.join();
        consume.join();
    }




    public static void delayQueueDemo(){
        ExecutorService executorService = Executors.newFixedThreadPool(3);

        class DelayPrint implements Delayed{

            private String name;

            private long scheduledTime;

            public DelayPrint(String name, long scheduledTime) {
                this.name = name;
                this.scheduledTime = scheduledTime;
            }

            @Override
            public long getDelay(TimeUnit unit) {
                return scheduledTime - System.currentTimeMillis();
            }

            @Override
            public int compareTo(Delayed o) {
                return Math.toIntExact(scheduledTime - ((DelayPrint) o).scheduledTime);
            }

            public void print(){
                System.out.printf("name:%s,scheduleTime:%d,startTime:%d,delay:%d%n",
                        name,
                        scheduledTime,
                        System.currentTimeMillis(),
                        scheduledTime-System.currentTimeMillis());
            }
        }

        DelayQueue<DelayPrint> delayPrints = new DelayQueue<>();

        for (int i = 0; i < 10; i++) {
            DelayPrint delayPrint = new DelayPrint("dp" + i, System.currentTimeMillis() + i * 2000);
            delayPrints.add(delayPrint);
            executorService.execute(()-> {
                try {
                    // 阻塞,直到时间到达
                    DelayPrint take = delayPrints.take();
                    take.print();
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        try {
            executorService.awaitTermination(1, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


    }



    static LinkedTransferQueue<String> lnkTransQueue = new LinkedTransferQueue<String>();

    /**
     * LinkedTransferQueue
     *
     */
    public static void linkedTransferQueue() {
        ExecutorService exService = Executors.newFixedThreadPool(2);
        Producer producer = new Producer();
        Consumer consumer = new Consumer();
        exService.execute(producer);
        exService.execute(consumer);
        exService.shutdown();
    }

    static class Producer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 3; i++) {
                try {
                    System.out.println("Producer is waiting to transfer...");
                    // 类似 SynchronousQueue, 阻塞传输对象,
                    // 必须有一个对应的take操作, transfer才能完成
                    lnkTransQueue.transfer("A" + i);
                    // 只是普通的放入队列操作, 不再说明
                    lnkTransQueue.put("A" + i);
                    System.out.println("producer transfered element: A" + i);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    static class Consumer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 6; i++) {
                try {
                    System.out.println("Consumer is waiting to take element...");
                    String s = lnkTransQueue.take();
                    System.out.println("Consumer received Element: " + s);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}


